package com.gis.aoitest.state;

import com.gis.aoitest.beans.SensorReading;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.*;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.Arrays;

/**
 * @author LnnuUser
 * @create 2021-09-08-下午8:11
 */
public class StateTest02_KeyedState {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // socket文本流
        DataStreamSource<String> inputStream = env.socketTextStream("localhost", 9999);

        // 转换成SensorReading类型
        DataStream<SensorReading> dataStream = inputStream.map(line -> {
            String[] fields = line.split(",");
            return new SensorReading(fields[0], new Long(fields[1]), new Double(fields[2]));
        } );

        SingleOutputStreamOperator<Integer> resultStream = dataStream
                .keyBy("id")
                .map(new MyKeyCountMapper());

        resultStream.print();

        env.execute();
    }

    // 自定义RichMapFunction
    public static class MyKeyCountMapper extends RichMapFunction<SensorReading, Integer> {

        private ValueState<Integer> keyCountState;

        // 其他类型状态的声明
        private ListState<String> myListState;
        private MapState<String, Double> myMapState;
        private ReducingState<SensorReading> myReducingState;

        @Override
        public void open(Configuration parameters) throws Exception {
            keyCountState = getRuntimeContext().getState(new ValueStateDescriptor<Integer>("keyCount", Integer.class, 0));

            myListState = getRuntimeContext().getListState(new ListStateDescriptor<String>("myList", String.class));

            myMapState = getRuntimeContext().getMapState(new MapStateDescriptor<String, Double>("myMap", String.class, Double.class));

//            myReducingState = getRuntimeContext().getReducingState(new ReducingStateDescriptor<SensorReading>())
        }

        @Override
        public Integer map(SensorReading sensorReading) throws Exception {
            Integer count = keyCountState.value();
            count++;
            keyCountState.update(count);
//            return count;

            // 其他状态api调用
            Iterable<String> strings = myListState.get();
            myListState.add("hello");
            myListState.addAll(Arrays.asList("hello"));
            for (String string : strings) {
                System.out.println(string);
            }

            // map state
            myMapState.get("1");
            myMapState.put("2", 12.3);

            // ReducingState
            myReducingState.add(sensorReading);

            myReducingState.clear();


            return count;
        }
    }
}
